/*
 * To change this template, choose Tools | Templates
 * and open the template in the editor.
 */
package com.snda.dw.hiveapi.hiveapitest;

import java.io.BufferedReader;
import java.io.IOException;
import java.util.Arrays;
import java.util.List;
import java.util.regex.Matcher;
import java.util.regex.Pattern;

/**
 *
 * @author lidejia
 */
public class HiveQueryResultParser {

    private Pattern jobNumberPattern = Pattern.compile("Total MapReduce jobs = (\\d+)");
    private Pattern startJobPattern = Pattern.compile("Launching Job");
    private Pattern jobProcessPattern = Pattern.compile("map = (\\d+)%,  reduce = (\\d+)%");
    private Pattern endJobPattern = Pattern.compile("Ended Job");
    private Pattern resultOKPattern = Pattern.compile("OK");
    private Pattern timeTakenPattern = Pattern.compile("Time taken: (\\d+)\\.(\\d+) seconds");
    private HiveSession session = null;

    public HiveQueryResultParser(HiveSession session) {
        this.session = session;
    }

    public HiveQueryResult parse(BufferedReader reader) {
        HiveQueryResult result = new HiveQueryResult();
        try {
            skipLines(3, reader);
            String line = null;
            Matcher matcher = null;
            line = reader.readLine();
            matcher = jobNumberPattern.matcher(line);
            if ((matcher.find())) {
                /**
                 * have map reduce job
                 */
                int jobNumber = Integer.parseInt(matcher.group(1));
                parseJobProcess(reader, jobNumber);
                line = reader.readLine();
                matcher = resultOKPattern.matcher(line);
                if (matcher.find()) {
                    parseResultData(reader, result);
                } else {
                    parseError(reader, line, result);
                }
            } else {
                session.setProcessPercentage(100);
                matcher = resultOKPattern.matcher(line);
                if (matcher.find()) {
                    /**
                     * no job
                     */
                    parseResultData(reader, result);
                } else {
                    /**
                     * error
                     */
                    parseError(reader, line, result);
                }
            }
        } catch (Exception e) {
            result.setErrorMessage("Internal Error");
        }
        return result;
    }

    private void parseJobProcess(BufferedReader reader, int jobNumber) throws IOException {
        Matcher matcher = null;
        String line = null;
        for (int i = 0; i < jobNumber; i++) {
            line = reader.readLine();
            matcher = startJobPattern.matcher(line);
            if (matcher.find()) {
                while (true) {
                    line = reader.readLine();
                    matcher = endJobPattern.matcher(line);
                    if (matcher.find()) {
                        break;
                    } else {
                        matcher = jobProcessPattern.matcher(line);
                        if (matcher.find()) {
                            int map = Integer.parseInt(matcher.group(1));
                            int reducer = Integer.parseInt(matcher.group(2));
                            int process = (int) (1.0 * (i * 200 + map + reducer) / (jobNumber * 2));
                            session.setProcessPercentage(process);
                        }
                    }
                }
            }
        }
    }

    private void parseResultData(BufferedReader reader, HiveQueryResult result) throws IOException {
        String line = null;
        // parse column headers
        line = reader.readLine();
        List<String> headers = Arrays.asList(line.split("\t"));
        result.addColumnNames(headers);
        // parse rows
        String lastLine = reader.readLine();
        while ((line = reader.readLine()) != null) {
            List<String> row = Arrays.asList(lastLine.split("\t"));
            result.addRow(row);
            lastLine = line;
        }
        // parse execute time
        Matcher matcher = timeTakenPattern.matcher(lastLine);
        if (matcher.find()) {
            String timePartOne = matcher.group(1);
            String timePartTwo = matcher.group(2);
            String time = timePartOne + "." + timePartTwo;
            double executeTimeInSecond = Double.parseDouble(time);
            result.setExecuteTimeInSecond(executeTimeInSecond);
        }
    }

    private void parseError(BufferedReader reader, String errorLine, HiveQueryResult result) throws IOException {
        StringBuilder sb = new StringBuilder(errorLine);
        String line = null;
        while ((line = reader.readLine()) != null) {
            sb.append(line);
        }
        result.setErrorMessage(sb.toString());
    }

    private void skipLines(int number, BufferedReader reader) throws IOException {
        for (int i = 0; i < number; i++) {
            reader.readLine();
        }
    }
}
